package com.shujia.mr.combine;


import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class CombineMapper extends Mapper<LongWritable,Text, Text, IntWritable> {
    /*
        写MapReduce代码处理时，需要先定义数据的处理逻辑

        编写流程：
            1.由于WordCountMapper需要符合MapReduce规范，所以需要先继承Mapper
                    Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
                        KEYIN, VALUEIN : 输入数据的泛型
                        KEYOUT, VALUEOUT : 输出数据的泛型

                    根据数据处理的逻辑
                        VALUEIN 就是一行文本数据 Java中的String类型 -Hadoop序列化-> Hadoop中的Text类型
                        KEYIN：是读取数据的偏移量（之前读取数据的总长度） Java中的long类型  -Hadoop序列化-> longWritable

                        KEYOUT: 是一个单词 对应是 Java中的String类型 -Hadoop序列化-> Hadoop中的Text类型
                        VALUEOUT 是一个标记值1 对应是 Java中的int类型 -Hadoop序列化-> IntWritable类型


          2.重写map函数
                在该函数中，可以定义MapTask中的数据处理逻辑
                       将一行单词进行按空格切分  ->  遍历每个单词  -> 拼接一个标记值1 -> 写出Key Value 数据到Reduce中
     */

    /**
     *
     * @param key 偏移量
     * @param value  一行文本数据
     * @param context 上下文对象，
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
//        super.map(key, value, context);

        String[] words = value.toString().split(" ");

        int flag = 1;
        for (String word : words) {
            // 在Map端中的context对象可以将数据写出到Reduce端
            //  write(KEYOUT var1, VALUEOUT var2)
            Text outKey = new Text(word);
            IntWritable outValue = new IntWritable(flag);
            context.write(outKey,outValue);
        }
    }
}
